package com.fy.service.server.consumer;

import com.fy.service.server.listener.BrokerMessageListener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Description：消费者配置,负责消费其他节点传递过来的message
 */
@Configuration
public class BrokerMessageConsumer {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private BrokerMessageListener brokerMessageListener;

    @Value("${spring.rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;

    @Value("${spring.rocketmq.consumer.groupName}")
    private String groupName;

    @Value("${spring.rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;

    @Value("${spring.rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;

    @Value("${spring.rocketmq.consumer.brokerMessageTopic}")
    private String brokerMessageTopic;

    @Bean(initMethod = "start", destroyMethod = "shutdown" ,name = "brokerMQPushConsumer")
    public DefaultMQPushConsumer getRocketMQConsumer() {


        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        //设置从队列头部开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //只消费一次
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setVipChannelEnabled(false);
        //设置一次性消费多少条
        consumer.setPullBatchSize(100);
        consumer.registerMessageListener(brokerMessageListener);
        try {
            consumer.subscribe(brokerMessageTopic,"*");
            logger.info("================>路由 消费者创建完成，ConsumerGroupName{}<================",groupName);
            return consumer;
        } catch (MQClientException e) {
            logger.error("消费者启动失败:" ,e);
        }
        return null;
    }

}
